package cn.ilikes.tools.nosql.jpa.clients.schemas;

import java.io.IOException;
import java.util.Map;
import java.util.Set;
import java.util.Map.Entry;

import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.TableNotFoundException;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.slf4j.Logger;

import cn.ilikes.tools.logs.Loggerfactory;
import cn.ilikes.tools.nosql.jpa.JPASQLPersistenUnitEnv;
import cn.ilikes.tools.nosql.jpa.SysConfigs;
import cn.ilikes.tools.nosql.jpa.clients.admins.IAdmin;
import cn.ilikes.tools.nosql.jpa.clients.admins.JPAHbaseAdmin;
import cn.ilikes.tools.nosql.jpa.metadatas.EntityMetaData;
import cn.ilikes.tools.nosql.jpa.metadatas.IEntityMetaDataFactory;
import cn.ilikes.tools.nosql.jpa.metadatas.elements.EntityFields;
import cn.ilikes.tools.nosql.jpa.unitls.JPAUtils;

public class HBaseSchemaManager extends SchemaManagerImples<HBaseAdmin> implements ISchemaManager {

	private static Logger logger = Loggerfactory.instance(HBaseSchemaManager.class);

	private ExecType execType;

	public HBaseSchemaManager(IAdmin<HBaseAdmin> admin) {
		super(admin);
	}

	public ExecType getExecType() {
		return execType;
	}

	public void setExecType(ExecType execType) {
		this.execType = execType;
	}

	@Override
	public void init() {
		IAdmin<HBaseAdmin> hbaseAdmin = getSchemaAdmin();
		hbaseAdmin.init();
		setInit(true);

	}

	@Override
	public void execs() throws JPASQLSchemaException {
		checkInit();
		Loggerfactory.info(logger, "HBase Schema Manager Start excec :");
		IAdmin<HBaseAdmin> hBaseAdmins = getSchemaAdmin();
		JPASQLPersistenUnitEnv persistenUnitEnv = hBaseAdmins.getHbasePersistenUnitEnv();
		String ddlAuto = (String) persistenUnitEnv.getPersistenceUnitMetadata().getProperties().get(SysConfigs.jpa_ddl_auto);
		ExecType execType = ExecType.valueOf(ddlAuto);
		if (execType.equals(ExecType.no))
			return;
		Loggerfactory.info(logger, "HBase Schema Manager Start Excec :" + ddlAuto);
		IEntityMetaDataFactory entityMetaDataFactory = persistenUnitEnv.getMetaDataManagerFactory();
		Set<EntityMetaData> entityMetaDatas = entityMetaDataFactory.listAllIEntityMetaData();

		for (EntityMetaData entityMetaData : entityMetaDatas) {
			HTableDescriptor tableDescriptor = getTableMetaData(entityMetaData);
			HBaseAdmin hBaseAdmin = hBaseAdmins.getAdmin();
			boolean result_ = false;
			try {
				result_ = hBaseAdmin.tableExists(tableDescriptor.getName());
				if (result_) {
					logger.info(" table " + entityMetaData.getEntityTableName() + "  exists ! ");
					try {
						hBaseAdmin.disableTable(tableDescriptor.getName());
						hBaseAdmin.deleteTable(tableDescriptor.getName());
					} catch (TableNotFoundException e) {
						e.printStackTrace();
						logger.info(" table " + entityMetaData.getEntityTableName() + " not  exists !   " + e.getMessage(), e.getMessage());
						//throw new JPASQLSchemaException(e);

					} catch (IOException ioex) {
						logger.error("Either table isn't in enabled state or some network problem, Caused by: " + ioex.getMessage());
						hBaseAdmin.deleteTable(tableDescriptor.getName());

					}
				}

				//				HTableDescriptor hTableDescriptor = getTableMetaData(entityMetaData);
				try {
					hBaseAdmin.createTable(tableDescriptor);
					logger.info(" create table " + entityMetaData.getEntityTableName() + "  success  ! ");
				} catch (IOException ioex1) {
					Loggerfactory.error(logger, "Table isn't in enabled state, Caused by:" + ioex1.getMessage(), ioex1);
				}
			} catch (IOException e1) {
				logger.info(" check  table " + entityMetaData.getEntityTableName() + "  is  exists  no success  ! " + e1.getMessage());
			}

		}

	}

	private HTableDescriptor getTableMetaData(EntityMetaData entityMetaData) {
		Loggerfactory.info(logger, "init   table  Info  descriptor   table name : " + entityMetaData.getEntityTableName());
		HTableDescriptor hTableDescriptor = new HTableDescriptor(entityMetaData.getEntityTableName());
		Set<Map.Entry<String, EntityFields>> entrySets = entityMetaData.getEntityFields().entrySet();
		for (Entry<String, EntityFields> entrySet : entrySets) {
			EntityFields entityFields = entrySet.getValue();
			{
				if (entityFields.getjPAField() == null || !entityFields.isSave())
					continue;
				String columnName = null;
				//if(Loggerfactory.isdebug(logger))Loggerfactory.debug(logger, 
				//		"table column 1 => columnName : "+entityFields.getFieldName() + "  = >  "+entityFields.getjPAField().isRelation());
				if (entityFields.getjPAField().isRelation()) {
					if (entityFields.getjPAField().getRelation().isColumnName())
						columnName = entityFields.getjPAField().getRelation().getColumnName();
				} else {
					columnName = entityFields.getjPAField().getColumnName();
				}
				//if(Loggerfactory.isdebug(logger))Loggerfactory.debug(logger, "table column 2=> columnName : "+entityFields.getFieldName() + "  = >  "+columnName);
				if (columnName == null || columnName.isEmpty())
					continue;
				int index = columnName.indexOf("@");
				String family = JPAUtils.getfamily(columnName, index);
				Loggerfactory.info(logger, "init table column => columnName : " + columnName + "  familyName : " + family);
				HColumnDescriptor hColumnDescriptor = new HColumnDescriptor(family);
				hTableDescriptor.addFamily(hColumnDescriptor);

			}
		}
		return hTableDescriptor;
	}

	public static void createSchema(JPASQLPersistenUnitEnv persistenUnitEnv) {
		Loggerfactory.info(logger, "Defult Schema Manager Factory   create " + persistenUnitEnv.getPersistenUnitName() + "  Hbase Schema  ");
		IAdmin<HBaseAdmin> admin = new JPAHbaseAdmin(persistenUnitEnv);
		ISchemaManager schemaManager = new HBaseSchemaManager(admin);
		try {
			schemaManager.init();
			schemaManager.execs();
		} catch (JPASQLSchemaException e) {
			Loggerfactory.error(logger, e);
		} finally {
			schemaManager.close();
		}
	}

	@Override
	public void close() {
		Loggerfactory.info(logger, "HBase Schema Manager Close ");
		IAdmin<HBaseAdmin> hbaseAdmin = getSchemaAdmin();
		hbaseAdmin.close();

	}
}
